import pandas as pd
from clickhouse_driver import Client
from datetime import datetime, timedelta
from tabulate import tabulate
from datetime import date
import logging
import time
from mysql_proxy import MySQLSSHClient
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ClickHouseDailyActive:
    def __init__(self, host='localhost', port=9000, user='default', 
                 password='', database='db_cfx'):
        self.client = Client(
            host=host,
            port=port,
            user=user,
            password=password,
            database=database,
            settings={'use_numpy': True}  # 可选，提高性能
        )
        self.database = database
    
    def get_daily_active_data(self, start_date, end_date=None):
        """
        获取指定日期范围的日活数据
        
        Args:
            start_date: 开始日期 (YYYY-MM-DD 或 datetime)
            end_date: 结束日期，默认为昨天
            
        Returns:
            DataFrame: 包含日活数据的DataFrame
        """
        if end_date is None:
            end_date = datetime.now() - timedelta(days=1)
        
        # 转换日期格式
        if isinstance(start_date, str):
            start_date = datetime.strptime(start_date, '%Y-%m-%d')
        if isinstance(end_date, str):
            end_date = datetime.strptime(end_date, '%Y-%m-%d')
        
        query = """
        SELECT 
            base_create_date,
            count(DISTINCT uuid) as daily_active_users,
            count(*) as total_events
        FROM t_log_base_data_standard_active
        WHERE base_create_date BETWEEN %(start_date)s AND %(end_date)s
        GROUP BY base_create_date
        ORDER BY base_create_date
        """
        
        params = {
            'start_date': start_date.strftime('%Y-%m-%d'),
            'end_date': end_date.strftime('%Y-%m-%d')
        }
        
        try:
            result = self.client.query_dataframe(query, params)
            logger.info(f"成功获取 {len(result)} 天的日活数据")
            return result
        except Exception as e:
            logger.error(f"查询日活数据失败: {e}")
            return pd.DataFrame()
    
    def get_daily_active_detail(self, target_date):
        """
        获取指定日期的详细日活数据
        """
        if isinstance(target_date, str):
            target_date = datetime.strptime(target_date, '%Y-%m-%d')
        
        query = """
        select activate_date,
        count(),
        uniqExact(oaid)
        from db_cfx.t_data_user_cpa_active 
        WHERE activate_date = %(target_date)s 
        group by activate_date
        
        """
        
        params = {'target_date': target_date.strftime('%Y-%m-%d')}
        
        try:
            result = self.client.query_dataframe(query, params)
            logger.info(f"成功获取 {target_date} 的详细日活数据，共 {len(result)} 条记录")
            return result
        except Exception as e:
            logger.error(f"查询详细日活数据失败: {e}")
            return pd.DataFrame()
    
    def check_data_completeness(self, target_date):
        """
        检查数据完整性（与目标表对比）
         source_anyselectedid_data AS (
            SELECT DISTINCT uuid
            FROM t_log_base_data_standard_active
            WHERE toDate(base_create_time) = %(target_date)s and base_app_id='AnySelectedID' and vn>'3.37'
        ),
        source_otrher_data AS (
            SELECT DISTINCT uuid
            FROM t_log_base_data_standard_active
            WHERE toDate(base_create_time) = %(target_date)s and base_app_id!='AnySelectedID'
        ),
                target_asid_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_active
            WHERE toDate(activate_time) = %(target_date)s and channel='AnySelectedID'
        ),
        target_other_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_active
            WHERE toDate(activate_time) = %(target_date)s and channel!='AnySelectedID'
        )
        """
        query = """
        WITH source_data AS (
            SELECT DISTINCT uuid
            FROM t_log_base_data_standard_active
            WHERE toDate(base_create_time) = %(target_date)s and ((base_app_id='AnySelectedID' and vn>'3.37') or base_app_id!='AnySelectedID')
        ),
        target_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_active
            WHERE toDate(activate_time) = %(target_date)s
        ),
        deduction_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_active_final
            WHERE toDate(activate_time) = %(target_date)s
        ),
        deduction_table_data AS (
            SELECT count(*) as deduction_total,
            count(CASE WHEN channel_name = 'lockscreen_false' THEN 1 END) AS deduction_lockscreen_false,
            count(CASE WHEN channel_name = '' THEN 1 END) AS deduction_hit
            FROM db_cfx.t_data_user_cpa_new_deduction
            WHERE toDate(activate_time) = %(target_date)s
        ),
        new_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_new
            WHERE toDate(activate_time) = %(target_date)s
        ),
        new_deduction_data AS (
            SELECT DISTINCT oaid
            FROM t_data_user_cpa_new_final
            WHERE toDate(activate_time) = %(target_date)s
        )

        SELECT 
            (SELECT count() FROM source_data) as expected_count,
            (SELECT count() FROM target_data) as actual_count,
            (SELECT count() FROM deduction_data) as dedution_actual_count,
            (SELECT deduction_total FROM deduction_table_data) as deduction_total,
            (SELECT deduction_lockscreen_false FROM deduction_table_data) as deduction_lockscreen_false,
            (SELECT deduction_hit FROM deduction_table_data) as deduction_hit,
            (SELECT count() FROM new_data) as new_count,
            (SELECT count() FROM new_deduction_data) as dedution_new_count
            
        """
        
        params = {'target_date': target_date.strftime('%Y-%m-%d')}
        
        try:
            result = self.client.query_dataframe(query, params)
            return result.iloc[0] if len(result) > 0 else None
        except Exception as e:
            logger.error(f"检查数据完整性失败: {e}")
            return None
        
    def get_mysql_active_data(self, target_date):
        """
        获取 MySQL 中的日活数据（示例）
        
        Args:
            date_str: 日期字符串 (YYYY-MM-DD)
            
        Returns:
            int: 日活用户数
        """
        print(f"获取 {target_date} 的 MySQL 日活数据")
        # 方法1: 常规使用
        client = MySQLSSHClient()
        if client.connect():
            try:
                data, columns = client.query(f"SELECT count(1) as cnt FROM data_user_cpa_daily_active_{target_date};")
                #logger.info(f"MySQL 日活用户数: {data[0][0]}")
                # for row in data:
                #     print(row)
                return data[0]['cnt']
            finally:
                client.close()

    def get_mysql_active_per_date(self, begin_date):
        """
        获取 MySQL 中的日活数据（示例）
        
        Args:
            date_str: 日期字符串 (YYYY-MM-DD)
            
        Returns:
            int: 日活用户数
        """
        dt = datetime.strptime(begin_date, "%Y-%m-%d")
            # 转换为UTC时间
        timestamp = int(time.mktime(dt.timetuple()))*1000
        print(f"获取 {begin_date} 的 MySQL 日活数据")
        # 方法1: 常规使用
        client = MySQLSSHClient()
        if client.connect():
            try:
                data, columns = client.query(f"SELECT date,SUM(dau) as dau,SUM(dnu) as dnu FROM cfx_t.data_cpa_user_dau WHERE date>={timestamp} group by date;")
                #logger.info(f"MySQL 日活用户数: {data[0][0]}")
                result=[]
                for row in data:
                    tmp=[]
                    tmp.append(row['date'])
                    tmp.append(int(row['dau']))
                    tmp.append(int(row['dnu']))
                    result.append(tmp)
                    print(row)
                return result
            finally:
                client.close()
                

    def get_mysql_active_deduction_date(self, begin_date):
        """
        获取 MySQL 中的日活数据（示例）
        
        Args:
            date_str: 日期字符串 (YYYY-MM-DD)
            
        Returns:
            int: 日活用户数
        """
        dt = datetime.strptime(begin_date, "%Y-%m-%d")
            # 转换为UTC时间
        timestamp = int(time.mktime(dt.timetuple()))*1000
        print(f"获取 {begin_date} 的 MySQL 日活数据")
        # 方法1: 常规使用
        client = MySQLSSHClient()
        if client.connect():
            try:
                data, columns = client.query(f"SELECT date,SUM(dau) as dau,SUM(dnu) as dnu FROM cfx_t.data_valid_user_dau WHERE date>={timestamp} group by date;")
                #logger.info(f"MySQL 日活用户数: {data[0][0]}")
                result=[]
                for row in data:
                    tmp=[]
                    tmp.append(row['date'])
                    tmp.append(int(row['dau']))
                    tmp.append(int(row['dnu']))
                    result.append(tmp)
                    print(row)
                return result
            finally:
                client.close()

    def get_mysql_deduction_date(self, begin_date):
        """
        获取 MySQL 中的日活数据（示例）
        
        Args:
            date_str: 日期字符串 (YYYY-MM-DD)
            
        Returns:
            int: 日活用户数
        """
        dt = datetime.strptime(begin_date, "%Y-%m-%d")
            # 转换为UTC时间
        timestamp = int(time.mktime(dt.timetuple()))*1000
        print(f"获取 {begin_date} 的 MySQL 日活数据")
        # 方法1: 常规使用
        client = MySQLSSHClient()
        if client.connect():
            try:
                data, columns = client.query(f"select DATE_FORMAT(FROM_UNIXTIME(activate_time/1000), '%Y-%m-%d') as date,COUNT(1) as cnt from cfx_t.data_user_device_valid_active where valid_type=0 and activate_time>={timestamp} GROUP BY DATE_FORMAT(FROM_UNIXTIME(activate_time/1000), '%Y-%m-%d');")
                #logger.info(f"MySQL 日活用户数: {data[0][0]}")
                result=[]
                for row in data:
                    tmp=[]
                    tmp.append(row['date'])
                    tmp.append(int(row['cnt']))
                    result.append(tmp)
                    print(row)
                return result
            finally:
                client.close()

    def display_date(self, headers,data):
        """展示同一日期不同维度的对比数据"""
        
        # 示例数据 - 同一日期的不同维度数据
        current_date = date.today().strftime("%Y-%m-%d")
        
        # 数据结构: [维度名称, 数值1, 数值2, 数值3, 变化率(%)]
        # data = [
        #     ["销售额", 15600, 14200, 13500, 15.56],
        #     ["访问量", 3240, 2890, 2650, 22.26],
        #     ["转化率", 4.8, 4.2, 3.9, 23.08],
        #     ["客单价", 286, 275, 260, 10.00],
        #     ["新用户数", 245, 198, 180, 36.11]
        # ]
        
        # 表格标题
        #print(f"===== {current_date} 各维度数据对比表 =====")
        # print("(数值1:今日, 数值2:昨日, 数值3:前日, 变化率:今日较前日)")
        # print()
        
        # 定义表头
        #headers = ["维度", "数值1", "数值2", "数值3", "变化率(%)"]
        
        # 打印表格
        print(tabulate(data, headers=headers, tablefmt="grid"))
        
        # 打印数据说明
        # print("\n数据说明:")
        # print("1. 变化率计算公式: (数值1 - 数值3) / 数值3 * 100%")
        # print("2. 正数表示增长，负数表示下降")

    

# 使用示例
if __name__ == "__main__":
    # 创建连接实例
    ch = ClickHouseDailyActive(
        host='172.29.151.181',  # 您的ClickHouse服务器IP
        port=9000,
        user='default',
        password='Keepgo123@cys',  # 如果有密码
        database='db_cfx'
    )
    
    # # 获取最近7天的日活数据
    # end_date = datetime.now() - timedelta(days=1)
    # start_date = end_date
    
    # daily_data = ch.get_daily_active_data(start_date, end_date)
    # print("日活数据概览:")
    # print(daily_data)
    
    # # # 获取某一天的详细数据
    # target_date = datetime.now() - timedelta(days=0)
    # detail_data = ch.get_daily_active_detail(target_date)
    # print(f"\n{target_date.strftime('%Y-%m-%d')} 详细数据:")
    # print(detail_data.head())
    query_days = 6
    begin_date = datetime.now() - timedelta(days=query_days)
    mysql_per_day = ch.get_mysql_active_per_date(begin_date.strftime('%Y-%m-%d'))
    mysql_active_deduction_day = ch.get_mysql_active_deduction_date(begin_date.strftime('%Y-%m-%d'))
    mysql_deduction_day = ch.get_mysql_deduction_date(begin_date.strftime('%Y-%m-%d'))
    for row in mysql_per_day:
        print(row)
        int_num = int(row[1])
        print( datetime.fromtimestamp(row[0]/1000),row[1],type(row[1]),type(int_num))
    data=[]
    new_data=[]
    index=0
    while query_days>=0:
        target_date = datetime.now() - timedelta(days=query_days)
        # 检查数据完整性
        completeness = ch.check_data_completeness(target_date)
        if completeness is not None:

            data_one_day =[]
            count_mysql=ch.get_mysql_active_data(target_date.day)
            if count_mysql==0:
                count_mysql=1
            print(f"count_mysql:{count_mysql}")
            data_one_day.append(target_date.strftime('%Y-%m-%d'))
           
            data_one_day.append(completeness['expected_count'])
            data_one_day.append(count_mysql)
            data_one_day.append(completeness['expected_count']-count_mysql)
            data_one_day.append(f"{(completeness['expected_count']-count_mysql)*100.0/count_mysql:.2f}%")

            data_one_day.append(completeness['actual_count'])
            data_one_day.append(mysql_per_day[index][1])
            data_one_day.append(completeness['actual_count']-mysql_per_day[index][1])
            data_one_day.append(f"{(completeness['actual_count']-mysql_per_day[index][1])*100.0/mysql_per_day[index][1]:.2f}%")

            data_one_day.append(completeness['dedution_actual_count'])
            data_one_day.append(mysql_active_deduction_day[index][1])
            data_one_day.append(completeness['dedution_actual_count']-mysql_active_deduction_day[index][1])
            data_one_day.append(f"{(completeness['dedution_actual_count']-mysql_active_deduction_day[index][1])*100.0/mysql_active_deduction_day[index][1]:.2f}%")

            data_one_day.append(completeness['actual_count']-completeness['dedution_actual_count'])
            data_one_day.append(f"{(completeness['actual_count']-completeness['dedution_actual_count'])*100.0/completeness['actual_count']:.2f}%")
            data_one_day.append(mysql_per_day[index][1]-mysql_active_deduction_day[index][1])
            data_one_day.append(f"{(mysql_per_day[index][1]-mysql_active_deduction_day[index][1])*100.0/mysql_per_day[index][1]:.2f}%")
            data.append(data_one_day)

            data_one_day_new =[]
            data_one_day_new.append(target_date.strftime('%Y-%m-%d'))
            data_one_day_new.append(completeness['new_count'])
            data_one_day_new.append(mysql_per_day[index][2])
            data_one_day_new.append(completeness['new_count']-mysql_per_day[index][2])
            data_one_day_new.append(f"{(completeness['new_count']-mysql_per_day[index][2])*100.0/mysql_per_day[index][2]:.2f}%")

            data_one_day_new.append(completeness['dedution_new_count'])
            data_one_day_new.append(mysql_active_deduction_day[index][2])
            data_one_day_new.append(completeness['dedution_new_count']-mysql_active_deduction_day[index][2])
            data_one_day_new.append(f"{(completeness['dedution_new_count']-mysql_active_deduction_day[index][2])*100.0/mysql_active_deduction_day[index][2]:.2f}%")

            data_one_day_new.append(completeness['new_count']-completeness['dedution_new_count'])
            data_one_day_new.append(f"{(completeness['new_count']-completeness['dedution_new_count'])*100.0/completeness['new_count']:.2f}%")
            data_one_day_new.append(mysql_per_day[index][2]-mysql_active_deduction_day[index][2])
            data_one_day_new.append(f"{(mysql_per_day[index][2]-mysql_active_deduction_day[index][2])*100.0/mysql_per_day[index][2]:.2f}%")
            data_one_day_new.append(completeness['deduction_total'])
            data_one_day_new.append(mysql_deduction_day[index][1])
            data_one_day_new.append(completeness['deduction_lockscreen_false'])
            data_one_day_new.append(completeness['deduction_hit'])
            new_data.append(data_one_day_new)

            
        index+=1
        
        # count = ch.get_mysql_active_data(target_date.day)
        # print(f"mysql日活数:{count}")
        # print(f"ch日志表完整率: {completeness['expected_count']*100/count:.2f}%")
        # print(f"ch日活表完整率: {completeness['actual_count']*100/count:.2f}%")
        # print(f"\n\n")
        query_days-=1
    print(f"===== {query_days} 每日活跃数据对比表 =====")
    print("+------------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+")
    print("+------------+--------------------原始日志--------------------+-----------------------活跃数据----------------+----------------------扣量后数据---------------+-----------------------扣量数据-----------------+")
    print("+-----日期---+-----ch---+----mysql---+----差值---+--比例(%)--+--活跃(ch)-+-活跃(mysql)+----差值--+--比例(%)--+-扣量后(ch)+扣量后(ms)-+----差值---+--比例(%)---+--扣量(ch)-+比例(ch)--+--扣量(ms)---+--比例(ms)--+")
    #headers = ["日期","mysql", "ch",  "差值", "比例(%)","活跃数","活跃数(扣)","扣量表","扣量数","比例(%)"]
    headers = ["-------","-------", "-------",  "-------", "-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------"]
    ch.display_date(headers,data)
    print("\n数据说明:")
    print("1. 差值: ch - mysql")
    print("2. 比例(%)计算公式: 差值 / mysql * 100%")
    print("")
    print(f"===== {query_days} 每日新增数据对比表 =====")
    print("+------------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+-----------+")
    print("+------------+-----------------------新增数据----------------+----------------------扣量后数据---------------+---------------------------扣量数据----------------+")
    print("+-----日期---+-新增(ch)-+-新增(ms)-+----差值--+--比例(%)--+-扣量后(ch)+扣量后(ms)-+----差值---+--比例(%)---+--扣量(ch)-+比例(ch)--+--扣量(ms)-+--比例(ms)--+扣量表(ch)-+扣量表(ms)-+扣量lock(ch)-+扣量hit(ch)-+")
    #headers = ["日期","mysql", "ch",  "差值", "比例(%)","活跃数","活跃数(扣)","扣量表","扣量数","比例(%)"]
    headers = ["-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------","-------"]
    ch.display_date(headers,new_data)
